package com.key.w8.http1.handler;

import com.key.w8.common.ChannelHandlerContextAware;
import com.key.w8.common.ConnectionAdapter;
import com.key.w8.common.HttpCommonHandler;
import com.key.w8.common.ReadOrWriteTimeHolder;
import com.key.w8.http1.connection.Connection;
import com.key.w8.http1.connection.ConnectionBuilder;
import com.key.w8.http1.connection.connectionImpl.DefaultHttp1Connection;
import com.key.w8.http1.constant.ExtendHttpResponseStatus;
import com.key.w8.http1.exception.Http1Exception;
import com.key.w8.http1.handler.codec.CodecBuilder;
import com.key.w8.http1.handler.codec.Http1Codec;
import com.key.w8.http1.handler.mockEntity.MockHttp1Request;
import com.key.w8.http1.handler.mockEntity.MockHttp1Response;
import io.netty.buffer.ByteBuf;
import io.netty.channel.*;
import io.netty.handler.codec.http.*;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.ReferenceCountUtil;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;

import static com.key.w8.http1.constant.Constants.*;

/**
 * @Author: k8
 * @CreateTime: 2025-02-19
 * @Version: 1.0
 */
@Slf4j
public class Http1FullHandler extends ChannelDuplexHandler implements HttpCommonHandler {
    /**
     * 在客户端 userHandler需要承担http请求的构建和http响应的处理
     * 在服务器端 userHandler需要承担http请求的处理和http响应的构建
     * 此外userHandler应该具备一些http请求的发送功能
     */
    private Http1Codec http1Codec;
    private CodecBuilder codecBuilder;
    private final boolean isServer;
    private boolean compressResponseContent;
    private long heartbeatInterval;
    private long connectionTimeout;
    /**
     * 一般ctx关闭就会关闭整个pipeline以及tcp连接
     * 现在将开端交给connection，其持有ctx，且可以拓展应用层关闭时的其他处理
     */
    private Connection connection;
    private ConnectionBuilder connectionBuilder;
    /**
     * 单位为字节
     */
    private int maxContentLength;

    private final Set<ChannelHandlerContextAware> contextAwareSet = new HashSet<>();
    private ReadOrWriteTimeHolder timeHolder;
    private static int DEFAULT_MAX_ERROR_TIMES = 16;
    //如果连续16次通讯都报异常则直接关闭，当读成功时还原
    private int maxErrorTimes = DEFAULT_MAX_ERROR_TIMES;
    boolean keepAlive;
    public Http1FullHandler(boolean isServer, long heartbeatInterval, long connectionTimeout,  int maxContentLength) {
        assert connectionTimeout > 0;
        assert heartbeatInterval > 0;
        assert heartbeatInterval < connectionTimeout;
        assert maxContentLength > 0;
        this.heartbeatInterval = heartbeatInterval;
        this.connectionTimeout = connectionTimeout;
        this.timeHolder = new ReadOrWriteTimeHolder(connectionTimeout);
        this.isServer = isServer;
        this.maxContentLength = maxContentLength;
        this.keepAlive = true;
    }
    public Http1FullHandler(boolean isServer,  int maxContentLength) {
        assert maxContentLength > 0;
        this.isServer = isServer;
        this.maxContentLength = maxContentLength;
        this.keepAlive = false;
    }


    /**
     * 在连接建立initChannel时会添加进一些handler，而在pipeline添加handler后，会调用handler的handlerAdded
     * 心跳检测机制：
     * 因为http1最好是客户端发送请求服务端响应，不同于http2服务器可以主动**推送**
     * 服务器定时检测是否有数据读，发送事件，下游处理器检测是由超时，超时则主动关闭连接
     * 客户端定时发送心跳检测报文，如果
     * @param ctx
     * @throws Exception
     */

    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        if (isServer){
            doHandlerAddedServer(ctx);
        }else {
            doHandlerAddedClient(ctx);
        }
        if (keepAlive){
            initHeartbeatHandler(ctx);
        }
    }

    /**
     * codec即保证了部分的简单编解码，又相当于一个pipeline到connection之间的转发器，
     * connection依赖于codec的写能力，而codec又需要把读到的数据转发给connection
     * 在connection关闭的同时，codec也需要进行关闭，而连带整个channel进行关闭
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        this.http1Codec = codecBuilder.build();
        if (isServer){
            this.connection = connectionBuilder.build();
        }
        http1Codec.setConnection(connection);
        if (connection instanceof DefaultHttp1Connection){
            DefaultHttp1Connection defaultHttp1Connection = (DefaultHttp1Connection) connection;
            defaultHttp1Connection.setHttp1Codec(http1Codec);
        }
        http1Codec.setTailChannelHandlerContext(ctx);
        connection.state(ConnectionAdapter.State.CREATED);
        connection.addCloseListener((connection)->{
            ctx.close();
        });
        super.channelActive(ctx);
    }

    @Override
    public void addChannelHandlerContextAware(ChannelHandlerContextAware aware) {
        contextAwareSet.add(aware);
    }

    private void doHandlerAddedClient(ChannelHandlerContext ctx) {
        ctx.pipeline()
                .addFirst(new HttpMockResolveHandler())
                .addFirst(new HttpContentDecompressor())
                .addFirst(new HttpObjectAggregator(maxContentLength))
                .addFirst(new BigHttpHandler())
                /*.addFirst(new ChannelInboundHandlerAdapter(){
                    @Override
                    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                        System.out.println("Response :");
                        System.out.println(msg.toString());
                        super.channelRead(ctx, msg);
                    }
                })*/
                .addFirst(new HttpClientCodec())
                /*.addFirst(new ChannelOutboundHandlerAdapter(){
                    @Override
                    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
                        if (msg instanceof ByteBuf) {
                            ByteBuf buf = (ByteBuf) msg;
                            byte[] bytes = new byte[buf.readableBytes()];
                            buf.getBytes(buf.readerIndex(), bytes);
                            System.out.println("Encoded HTTP Request:\n" + new String(bytes));
                        }
                        ctx.write(msg, promise); // 继续传递原始数据
                    }
                })*/;
    }


    /**
     * todo 需要修改，单独用一个handler聚合服务器端的处理器，因为协议协商还需要单独加处理器
     * 添加顺序原因是将这些处理器放到this的前面，这样在最后发送http的心跳包时就可以被处理成功
     *
     * @param ctx
     */
    private void doHandlerAddedServer(ChannelHandlerContext ctx) {
        ctx.pipeline().addFirst(new HttpMockResolveHandler());
        if (compressResponseContent){
            ctx.pipeline().addFirst(new HttpContentCompressor());
        }
        ctx.pipeline()
                .addFirst(new HttpObjectAggregator(maxContentLength))//获取完整的FullRequestDecoder
                .addFirst(new BigHttpHandler())
                .addFirst(new HttpServerCodec())//即使对压缩后的FullHttpResponse编码成字节流也可能对响应体进行拆解
                ;
    }
    /**
     * 拦截非法请求
     * 事件不能从后向前传递，又不好直接索要某个handler，故此使用监听器执行回调的方式
     * @param ctx
     * @param msg
     * @throws Exception
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        if (msg instanceof FullHttpResponse){
            FullHttpResponse response = (FullHttpResponse) msg;
            try{
                HttpHeaders headers = response.headers();
                if (!headers.contains(PONG_HEADER_NAME)) {//执行回调
                    http1Codec.responseRead(ctx,response);
                }else {
                    //log.debug("---------------------------pong");
                }
                maxErrorTimes = DEFAULT_MAX_ERROR_TIMES;
            }catch (Exception e){
                throw e;
            }finally {
                //当无法处理数据类型的时候就不会处理ByteBuf，就需要在这里释放
                ReferenceCountUtil.release(msg);
            }
        }else if (msg instanceof FullHttpRequest){
            FullHttpRequest request = (FullHttpRequest) msg;
            try{
                HttpHeaders headers = request.headers();
                if (headers.contains(PING_HEADER_NAME) && headers.contains(Time_Header)){
                    //log.debug("------------------------------ping");
                    //pong
                    http1Codec.pong();
                }else {
                    http1Codec.requestRead(ctx,request);
                }
                maxErrorTimes = DEFAULT_MAX_ERROR_TIMES;
            }catch (Exception e){
                throw e;
            }finally {
                ReferenceCountUtil.release(msg);
            }
        }else {
            super.channelRead(ctx,msg);
        }
    }

    public void setCodecBuilder(CodecBuilder codecBuilder) {
        this.codecBuilder = codecBuilder;
    }

    /**
     * HttpObjectAggregator对头部标注超过最大长度的消息是不会处理的，但是不能直接把头部给拦截掉，否则还是会对后续content进行放行
     * 则没有handler去处理这些content的ByteBuf
     * 故此需要有一个mock伪装HttpObject，对于超过大小的HttpObject，需要自己去生成一个伪装的HttpResponse或request
     * HttpObjectAggregator不会去处理它，但是后续处理器可以去处理
     * 故此还需要有处理器包装一层
     */
    private class HttpMockResolveHandler extends ChannelInboundHandlerAdapter{
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {

            if (!isServer){
                if (msg instanceof FullHttpResponse){
                    FullHttpResponse response = (FullHttpResponse) msg;
                    super.channelRead(ctx,response);
                }else if (msg instanceof MockHttp1Response){
                    MockHttp1Response response = (MockHttp1Response) msg;
                    super.channelRead(ctx,response.getHttpResponse());
                }
            }else {
                if (msg instanceof FullHttpRequest){
                    FullHttpRequest request = (FullHttpRequest) msg;
                    super.channelRead(ctx,request);
                }else if (msg instanceof  MockHttp1Request){
                    MockHttp1Request request = (MockHttp1Request) msg;
                    super.channelRead(ctx,request.getHttpRequest());
                }
            }
        }
    }
    private class BigHttpHandler extends ChannelInboundHandlerAdapter{
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            try {
                if (!isServer && msg instanceof HttpResponse){
                    HttpResponse response = (HttpResponse) msg;
                    HttpHeaders headers = response.headers();
                    int contentLength = Integer.parseInt(headers.get(HttpHeaderNames.CONTENT_LENGTH, "0"));
                    if (contentLength > maxContentLength){
                        //不同于超大请求，如果响应过大显然是要给后面的处理器提示的，说明响应过大
                        //需要传递一个HttpObjectAggregator无法处理但userHandler可以处理的响应
                        HttpResponse largeResponse = getLargeResponse();
                        HttpHeaders largeHeaders = largeResponse.headers();
                        for (Map.Entry<String, String> header : response.headers()) {
                            largeHeaders.set(header.getKey(),header.getValue());
                        }
                        largeResponse.headers().set(HttpHeaderNames.CONTENT_LENGTH,0);
                        MockHttp1Response mockHttp1Response = new MockHttp1Response(largeResponse);
                        //交给用户处理器处理伪装标记了过大的响应
                        super.channelRead(ctx, mockHttp1Response);
                        //交给HttpObjectAggregator处理完整的超大响应,会被丢弃
                        super.channelRead(ctx,response);
                    }else {
                        super.channelRead(ctx,msg);
                    }
                }else if (isServer && msg instanceof HttpRequest){
                    HttpRequest request = (HttpRequest) msg;
                    HttpHeaders headers = request.headers();
                    int contentLength = Integer.parseInt(headers.get(HttpHeaderNames.CONTENT_LENGTH, "0"));
                    if (contentLength > maxContentLength){
                        HttpRequest largeRequest = getLargeRequest();
                        HttpHeaders largeHeaders = largeRequest.headers();
                        for (Map.Entry<String, String> header : request.headers()) {
                            largeHeaders.set(header.getKey(),header.getValue());
                        }
                        largeHeaders.set(HttpHeaderNames.CONTENT_LENGTH,0);
                        MockHttp1Request mockHttp1Request = new MockHttp1Request(largeRequest);
                        super.channelRead(ctx, mockHttp1Request);
                        //还是需要交给HttpObjectAggregator处理ByteBuf，但是由于HttpObjectAggregator会丢弃该消息
                        //所以响应需要此处给
                        super.channelRead(ctx,request);
                    }else {
                        super.channelRead(ctx,msg);
                    }
                }else {
                    super.channelRead(ctx,msg);
                }
            }catch (Throwable throwable){
                throw new Http1Exception(throwable);
            }
        }

        private HttpRequest getLargeRequest() {
            HttpRequest request = new DefaultFullHttpRequest(HttpVersion.HTTP_1_1,HttpMethod.GET,"");
            request.headers().set(REQUEST_ENTITY_LARGE,true);
            return request;
        }

        private HttpResponse getLargeResponse(){
            HttpResponse response = new DefaultFullHttpResponse(
                    HttpVersion.HTTP_1_1, ExtendHttpResponseStatus.RESPONSE_ENTITY_TOO_LARGE);
            return response;
        }
    }
    /**
     * 替换成时间轮执行，并对初次检测时间做修改，尽量不要让检测时间太过分散
     * @param ctx
     */
    @Override
    public void initHeartbeatHandler(ChannelHandlerContext ctx) {
        ctx.pipeline().addFirst(timeHolder);
        if (!isServer){
            ctx.channel().pipeline().addFirst(
                    new IdleStateHandler(0, 0, heartbeatInterval, TimeUnit.MILLISECONDS));
        }else {
            ctx.channel().pipeline().addFirst(
                    new IdleStateHandler(0, 0,connectionTimeout+1, TimeUnit.MILLISECONDS));
        }
    }

    /**
     * 如果heartbeatInterval时间内没有
     * @param ctx
     * @param evt
     * @throws Exception
     */
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent && ((IdleStateEvent) evt).state() == IdleState.ALL_IDLE) {
            if (timeHolder.isTimeout()) {
                connection.close();
            }else if (!isServer){
                http1Codec.ping();
            }
        }else {
            super.userEventTriggered(ctx,evt);
        }
    }

    /**
     * 统一的异常收集，当异常数超过最大异常接收数则主动关闭
     * @param ctx
     * @param cause
     * @throws Exception
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        if (maxErrorTimes > 1){
            maxErrorTimes--;
            super.exceptionCaught(ctx,cause);
        }else {
            connection.close();
        }
    }

    /**
     * 检测是否是主动关闭，如果是connection的状态则不是created
     * 被动关闭就尝试去关闭connection，所以connection需要满足支持竞争关闭的情况
     * 因为eventLoop和主动关闭的线程可能不是同一个线程
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        if (connection.state().value == ConnectionAdapter.State.CREATED.value){//被动关闭
            connection.tryClose();//被动关闭则尝试tryClose，对于服务端来说没有支持重连，最后会直接关闭
        }
    }


    public void setCompressResponseContent(boolean compress){
        this.compressResponseContent = compress;
    }

    public Connection getConnection() {
        return connection;
    }

    public void setConnection(Connection connection) {
        if (isServer) throw new IllegalStateException("Just client need init connection.");
        this.connection = connection;
    }

    public void setConnectionBuilder(ConnectionBuilder connectionBuilder) {
        if (!isServer) throw new IllegalStateException("Just server need init connection builder.");
        this.connectionBuilder = connectionBuilder;
    }

}
